package com.chinamobile.websocket.websocketdispatcher.distributor.kafka;

import com.alibaba.fastjson.JSON;
import com.chinamobile.websocket.domain.bo.WebsocketKafkaMessageBo;
import com.chinamobile.websocket.websocketdispatcher.common.MessageDistributorConstants;
import com.chinamobile.websocket.websocketdispatcher.distributor.AbstractMessageDistributor;
import com.chinamobile.websocket.domain.dto.MessageDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @className: KafkaMessageDistributor
 * @description 基于 kafka 的消息分发器, 订阅 websocket 发送消息，接收到消息时进行推送
 * @author: fanzhijun
 * @date: 2024/1/22
 **/

@Slf4j
public class KafkaMessageDistributor<T extends MessageDto> extends AbstractMessageDistributor<T> {
    private final KafkaTemplate<String, WebsocketKafkaMessageBo> kafkaTemplate;

    public KafkaMessageDistributor(KafkaTemplate<String, WebsocketKafkaMessageBo> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * @title 消息生产
     * @author fanzhijun
     * @updateTime 2024/1/22 10:08
     */
    @Override
    public void distribute(T t) {
        WebsocketKafkaMessageBo kafkaMessageBo = new WebsocketKafkaMessageBo();
        kafkaMessageBo.setData(t.getData());
        kafkaMessageBo.setUniKey(t.getUniKey());
        kafkaMessageBo.setType(t.getType());
        kafkaTemplate.send(MessageDistributorConstants.KAFKA_WEBSOCKET_MESSAGE_TOPIC, kafkaMessageBo).addCallback(new ListenableFutureCallback<SendResult<String, WebsocketKafkaMessageBo>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.debug("发送消息失败.", throwable);
            }

            @Override
            public void onSuccess(SendResult<String, WebsocketKafkaMessageBo> stringKafkaMessageBoSendResult) {
                log.debug("发送消息成功." + JSON.toJSONString(stringKafkaMessageBoSendResult));
            }
        });
    }

}
